package com.cg.server;

import com.cg.entity.Message;
import com.cg.utils.MessageUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Component
@ServerEndpoint("/chat/{userId}")
@Slf4j
public class WebSocketServer {

    //用来存储每个用户客户端对象的WebSocketServer对象
    private static final ConcurrentHashMap<String, WebSocketServer> WEB_SOCKET_MAP = new ConcurrentHashMap<>();

    //实例一个session，这个session是websocket的session
    private Session session;

    //模拟当前登录的用户id
    private Integer userId;

    /**
     * 请求websocket处理事件
     */
    @OnOpen
    public void onOpen(@PathParam("userId") Integer userId, Session session) {
        log.info("【请求websocket处理事件,userId:{}】", userId);
        this.session = session;

        //新建连接后存储当前对应实例
        WEB_SOCKET_MAP.put(userId.toString(), this);

        //保存当前模拟登录的userId
        this.userId = userId;

        //系统广播通知（如: xxx上线了）
        String message = MessageUtils.getMessage(true, null, null, "用户{" + userId + "}已上线，当前在线人数: " + WEB_SOCKET_MAP.size());
        this.broadcastAllUsers(message, userId);
    }

    /**
     * 关闭websocket连接处理事件
     */
    @OnClose
    public void onClose(@PathParam("userId") String userId) {
        log.info("【websocket连接断开,userId:{}】", userId);
        WEB_SOCKET_MAP.remove(userId);
    }

    /**
     * 向服务端发送消息的处理事件
     */
    @OnMessage
    public void onMessage(String message, Session session) throws IOException {
        log.info("【websocket消息体,message:{}】", message);

        ObjectMapper mapper = new ObjectMapper();
        Message msg = mapper.readValue(message, Message.class);

        String msgStr = MessageUtils.getMessage(false, userId.toString(), msg.getToUserId(), msg.getContent());
        broadcastUser(msg.getToUserId(), msgStr);
    }

    /**
     * websocket连接异常处理时间
     */
    @OnError
    public void onError(Throwable throwable) {
        log.error("连接断开,错误信息：{}", throwable.getMessage());
    }

    public void broadcastAllUsers(String message, Integer userId) {
        WEB_SOCKET_MAP.values().stream()
                .filter(o -> !Objects.equals(o.userId, userId)) //不通知自己
                .collect(Collectors.toSet())
                .forEach(o -> {
                    try {
                        o.session.getBasicRemote().sendText(message);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
    }

    public void broadcastUser(String userId, String message) throws IOException {
        WebSocketServer webSocketServer = Optional.ofNullable(WEB_SOCKET_MAP.get(userId))
                .orElseThrow(() -> new RuntimeException("当前WebSocket实例不存在"));
        webSocketServer.session.getBasicRemote().sendText(message);
    }

}
